package com.lagou.filter;

import org.apache.commons.lang3.concurrent.ConcurrentUtils;
import org.apache.commons.lang3.time.StopWatch;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.common.extension.SPI;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.apache.dubbo.rpc.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * @author xingyi
 */
@Activate(group = CommonConstants.CONSUMER)
public class TPMonitorFilter implements Filter {

    private static final Logger logger = LoggerFactory.getLogger(TPMonitorFilter.class);

    private ScheduledExecutorService monitor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("tp-monitor"));
    private ScheduledExecutorService reseter = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("tp-reseter"));
    private ConcurrentHashMap<String, List<Long>> ellapseMap = new ConcurrentHashMap<>();

    public TPMonitorFilter() {
        monitor.scheduleAtFixedRate(() -> {
            for (Map.Entry<String, List<Long>> entry : ellapseMap.entrySet()) {
                String key = entry.getKey();
                List<Long> ellapseList = entry.getValue();
                if (ellapseList.isEmpty()) {
                    logger.info("{} monitor is empty", key);
                    continue;
                }
                ellapseList.sort(Long::compareTo);
                int t90Index = (int) (ellapseList.size() * 0.9);
                int t99Index = (int) (ellapseList.size() * 0.99);
                long t90 = ellapseList.get(t90Index);
                long t99 = ellapseList.get(t99Index);
                logger.info("{} t90:{}", key, t90);
                logger.info("{} t99:{}", key, t99);
            }
        }, 0, 5, TimeUnit.SECONDS);

        reseter.scheduleAtFixedRate(() -> {
            for (List<Long> ellapseList : ellapseMap.values()) {
                ellapseList.clear();
            }
        }, 60, 60, TimeUnit.SECONDS);
    }

    @Override
    public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        try {
            return invoker.invoke(invocation);
        } finally {
            stopWatch.stop();
            long ellapse = stopWatch.getTime();
            List<Long> ellapseList = ConcurrentUtils.putIfAbsent(ellapseMap, invocation.getServiceName() + '.' + invocation.getMethodName(), Collections.synchronizedList(new ArrayList<>()));
            ellapseList.add(ellapse);
        }
    }
}
